package com.yumeko.consumer;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class MySimpleConsumer3 {
    private static final Logger log = LoggerFactory.getLogger(MySimpleConsumer3.class);

    public static void main(String[] args) throws ClientException, IOException {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();

        String endpoints = "192.168.88.132:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder().setEndpoints(endpoints).build();
        String consumerGroup = "yourConsumerGroup";
        String tag = "*"; //
        String topic = "TransactionTopic";

        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setAwaitDuration(Duration.ofSeconds(3))
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();
        int maxMessageNum = 16;
        Duration invisibleDuration = Duration.ofSeconds(10);

        try {
            do {
                List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
                log.info("Received {} message(s)", messages.size());

                for (MessageView message : messages) {
                    final MessageId messageId = message.getMessageId();
                    try {
                        System.out.println("Consume Fifo message successfully, messageId:"+messageId);
                    } catch (Exception e) {
                        log.error("Failed to process message, messageId={}", messageId, e);
                    }
                }
            } while (true);
        } finally {
            consumer.close();
        }
    }
}
